/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.ban.data.exchange.runner.master;

import cn.ac.iie.di.ban.data.exchange.runner.commons.protocol.ScopeEnum;
import cn.ac.iie.di.ban.data.exchange.runner.commons.protocol.StatusEnum;
import cn.ac.iie.di.ban.data.exchange.runner.commons.protocol.TaskTypeEnum;
import cn.ac.iie.di.ban.data.exchange.runner.commons.protocol.proxy.DataProxyRequest;
import cn.ac.iie.di.ban.data.exchange.runner.commons.protocol.proxy.DataProxyResponse;
import cn.ac.iie.di.ban.data.exchange.runner.commons.protocol.task.order.TaskFinishOrderRequest;
import cn.ac.iie.di.ban.data.exchange.runner.commons.protocol.task.order.TaskOrderRequest;
import cn.ac.iie.di.ban.data.exchange.runner.commons.protocol.task.order.TaskOrderResponse;
import cn.ac.iie.di.ban.data.exchange.runner.commons.protocol.task.stat.TaskInfo;
import cn.ac.iie.di.ban.data.exchange.runner.commons.worker.MasterWorker;
import cn.ac.iie.di.ban.data.exchange.runner.master.helper.utils.utils;
import com.alibaba.fastjson.JSONObject;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.function.Function;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.log4j.Logger;

/**
 * @author root
 */
public class QueryHandler implements MasterWorker {

    private static String type = "(3)";

    static Logger logger = Logger.getLogger(QueryHandler.class);
    private final SimpleDateFormat TIMEFORMAT = new SimpleDateFormat("yyyyMMddHHmmss");
    private List<String> metadataUrls;
    private Function<TaskOrderRequest, CompletableFuture<TaskOrderResponse>> producer;
    private String hdfsPathPrefix;
    private Configuration hdfsConf;
    private List<String> websiteUrls;
    private int queryMaxNum;

    public QueryHandler(List<String> metadataUrls, String hdfsPathPrefix, Function<TaskOrderRequest, CompletableFuture<TaskOrderResponse>> producer, Configuration hdfsConf, List<String> websiteUrls, int queryMaxNum) {
        this.metadataUrls = metadataUrls;
        this.producer = producer;
        this.hdfsPathPrefix = hdfsPathPrefix;
        this.hdfsConf = hdfsConf;
        this.websiteUrls = websiteUrls;
        this.queryMaxNum = queryMaxNum;
    }

    @Override
    public DataProxyResponse execute(DataProxyRequest workerInfo) {

        SimpleDateFormat dateFormat = new SimpleDateFormat("yyyyMMdd");
        String dateStr = dateFormat.format(new Date(System.currentTimeMillis()));
        String timestamp = TIMEFORMAT.format(new Date(System.currentTimeMillis()));
        DataProxyResponse finishResponse;
        String returnMessage;
        String taskHdfsResult;
        int count = RequestCount.addUp();
        String uniqSql = "";
        String taskResult = "";
        String reqId;
        String taskId;
        String srcTopic;
        String uid = workerInfo.getUid();
        String sql = workerInfo.getSql();
        ScopeEnum scopeEnum = workerInfo.getScope();
        int scope = scopeEnum.getValue();
        String uniqKey = workerInfo.getUniq_key();
        reqId = "req_" + scope + "_" + timestamp + "_" + count;
        taskId = "task_" + scope + "_" + timestamp + "_" + count;
        Boolean queryTaskFull;

        logger.info(taskId + " the uid is " + uid + " ,sql is " + sql + " ,scope is " + scope + " ,uniq_key is " + uniqKey + " ,req_id is " + reqId);

        utils utils = new utils();
        try {
            queryTaskFull = utils.checkTaskFull(uid, websiteUrls, metadataUrls, queryMaxNum, type);
        } catch (Exception e) {
            logger.error(ExceptionUtils.getFullStackTrace(e));
            return new DataProxyResponse(uniqKey, reqId, StatusEnum.QUERY_TASK_ERROR, ExceptionUtils.getFullStackTrace(e));
        }
        if (queryTaskFull) {
            logger.error("query task is full. ");
            try {
                String status = "5";
                String detail = "query task is full. ";
                utils.sendRequestNew(reqId, scope, sql, status, detail, taskResult, uniqSql, taskId, uniqKey, uid, metadataUrls);
            } catch (Exception e) {
                logger.error(ExceptionUtils.getFullStackTrace(e));
                return new DataProxyResponse(uniqKey, reqId, StatusEnum.SUBSCRIPTION_TASK_ERROR, ExceptionUtils.getFullStackTrace(e));
            }
            return new DataProxyResponse(uniqKey, reqId, StatusEnum.QUERY_TASK_ERROR, "query task is full");
        }


        try {
            srcTopic = TopicParser.getSrcTopicName(sql);
        } catch (Exception e) {
            srcTopic = "default";
            logger.error(taskId + " parser sql get table failed, cause " + e.getMessage(), e);
        }
        taskResult = srcTopic + "_" + timestamp + "_" + count;
        logger.info(taskId + " req_id is " + reqId + " ,task_id is " + taskId + " ,result is " + taskResult);
        taskHdfsResult = hdfsPathPrefix + "/" + dateStr + "/" + taskResult;

        //2分派认证任务给slave
        HashMap<String, Object> salveMap = new HashMap<>();
        salveMap.put("sql", sql);
        salveMap.put("scope", scopeEnum);
        salveMap.put("uniq_key", uniqKey);
        salveMap.put("type", TaskTypeEnum.CHECK);
        salveMap.put("result_path", taskHdfsResult);
        salveMap.put("task_id", taskId);
        salveMap.put("uid", uid);
        TaskOrderResponse authResponse = SlaveTools.sendToSlave(producer, salveMap);
        StatusEnum authEnum = authResponse.getStatus();
        uniqSql = authResponse.getDetail();
        returnMessage = uniqSql;
        logger.info(taskId + " request slave to identity authentication, and return " + authEnum);
        if (authEnum.equals(StatusEnum.CHECK_SUCCESS)) {
            logger.info(taskId + " this user has permissions, and the only authentication sql is " + uniqSql);
            //查询元数据库中是否有对应的task_id，是否已经做过相同的任务
            HashMap<String, String> requestMap = new HashMap<>();
            requestMap.put("scope", "request");
            HashMap<String, String> metadataMap = new HashMap<>();
            metadataMap.put("scope", "task");
            //没有执行过相同的任务
            //5.2发送初始化状态至元数据服务
            requestMap.put("opration", "new");
            requestMap.put("condition", "");
            Map<String, Object> initRequestContent = new HashMap<>();
            initRequestContent.put("req_id", reqId);
            initRequestContent.put("type", scope);
            initRequestContent.put("usql", sql);
            initRequestContent.put("status", "1");
            initRequestContent.put("uniq_seq", uniqSql);
            initRequestContent.put("task_id", taskId);
            initRequestContent.put("uid", uid);
            initRequestContent.put("start", System.currentTimeMillis() / 1000);
            requestMap.put("content", new JSONObject(initRequestContent).toString());
            String requestInitReturn = ParserFastJson.requestMetadata(metadataUrls, requestMap);
            logger.info(taskId + " init new request, and return " + requestInitReturn);
            boolean requestInitFlag = ParserFastJson.returnError(requestInitReturn);
            if (requestInitFlag) {
                return new DataProxyResponse(uniqKey, reqId, StatusEnum.QUERY_TASK_ERROR, requestInitReturn);
            }

            metadataMap.put("opration", "new");
            metadataMap.put("condition", "");
            Map<String, Object> initContent = new HashMap<>();
            initContent.put("task_id", taskId);
            initContent.put("type", scope);
            initContent.put("uniq_sql", uniqSql);
            initContent.put("req_id", reqId);
            initContent.put("status", "1");
            initContent.put("worker", uid);
            initContent.put("start", System.currentTimeMillis() / 1000);
            metadataMap.put("content", new JSONObject(initContent).toString());
            String initReturn = ParserFastJson.requestMetadata(metadataUrls, metadataMap);
            logger.info(taskId + " init new task, and return " + initReturn);
            boolean initFlag = ParserFastJson.returnError(initReturn);
            if (initFlag) {
                return new DataProxyResponse(uniqKey, reqId, StatusEnum.QUERY_TASK_ERROR, initReturn);
            }

            Path hdfspath = new Path(taskHdfsResult);
            FileSystem fs = HDFSUtil.getHDFS(hdfsConf, hdfspath.toString());
            int trynum = 0;
            while (trynum < 3) {
                try {
                    trynum++;
                    fs.delete(hdfspath, true);
                    fs.mkdirs(hdfspath);
                    logger.info(taskId + " create hdfs directory finish.");
                    break;
                } catch (Exception ex) {
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException ex1) {
                    }
                    if (trynum == 3) {
                        logger.error(taskId + " create hdfs direcoty failed, cause " + ex.getMessage(), ex);
                    }
                }
            }

            //7更新元数据库对应result字段
            requestMap.put("opration", "update");
            Map<String, Object> updateRequestContent = new HashMap<>();
            updateRequestContent.put("result", taskHdfsResult);
            updateRequestContent.put("status", "2");
            requestMap.put("content", new JSONObject(updateRequestContent).toString());
            Map<String, Object> updateRequestCondition = new HashMap<>();
            updateRequestCondition.put("req_id", reqId);
            requestMap.put("condition", new JSONObject(updateRequestCondition).toString());
            String updateRerquestReturn = ParserFastJson.requestMetadata(metadataUrls, requestMap);
            logger.info(taskId + " update result and status of request, and return " + updateRerquestReturn);
            boolean updateRerquestFlag = ParserFastJson.returnError(updateRerquestReturn);
            if (updateRerquestFlag) {
                return new DataProxyResponse(uniqKey, reqId, StatusEnum.QUERY_TASK_ERROR, updateRerquestReturn);
            }

            metadataMap.put("opration", "update");
            Map<String, Object> updateContent = new HashMap<>();
            updateContent.put("result", taskHdfsResult);
            updateContent.put("status", "2");
            metadataMap.put("content", new JSONObject(updateContent).toString());
            Map<String, Object> updateCondition = new HashMap<>();
            updateCondition.put("task_id", taskId);
            metadataMap.put("condition", new JSONObject(updateCondition).toString());
            String updateReturn = ParserFastJson.requestMetadata(metadataUrls, metadataMap);
            logger.info(taskId + " update result and status of task, and return " + updateReturn);
            boolean updateFlag = ParserFastJson.returnError(updateReturn);
            if (updateFlag) {
                return new DataProxyResponse(uniqKey, reqId, StatusEnum.QUERY_TASK_ERROR, updateReturn);
            }

            //9给slave分派任务
            salveMap.put("type", TaskTypeEnum.RUN);
            TaskOrderResponse slaveResponse = SlaveTools.sendToSlave(producer, salveMap);
            StatusEnum slaveEnum = slaveResponse.getStatus();
            logger.info(taskId + " allocating task to salve, and return " + slaveEnum);
            if (slaveEnum.equals(StatusEnum.RUNNING)) {
                //10更新任务状态
                Map<String, Object> updateRequestStatus = new HashMap<>();
                updateRequestStatus.put("status", "3");
                updateRequestStatus.put("detail", "running");
                requestMap.put("content", new JSONObject(updateRequestStatus).toString());
                String updateRequestStatusReturn = ParserFastJson.requestMetadata(metadataUrls, requestMap);
                logger.info(taskId + " slave begin to run, update status of request, and return " + updateRequestStatusReturn);
                boolean updateRequestStatusFlag = ParserFastJson.returnError(updateRequestStatusReturn);
                if (updateRequestStatusFlag) {
                    return new DataProxyResponse(uniqKey, reqId, StatusEnum.QUERY_TASK_ERROR, updateRequestStatusReturn);
                }

                Map<String, Object> updateStatus = new HashMap<>();
                updateStatus.put("status", "3");
                updateStatus.put("detail", "running");
                metadataMap.put("content", new JSONObject(updateStatus).toString());
                String updateStatusReturn = ParserFastJson.requestMetadata(metadataUrls, metadataMap);
                logger.info(taskId + " slave begin to run, update status of task, and return " + updateStatusReturn);
                boolean updateStatusFlag = ParserFastJson.returnError(updateStatusReturn);
                if (updateStatusFlag) {
                    return new DataProxyResponse(uniqKey, reqId, StatusEnum.QUERY_TASK_ERROR, updateStatusReturn);
                }
            } else {
                Map<String, Object> finishRequestContent = new HashMap<>();
                finishRequestContent.put("status", "5");
                finishRequestContent.put("detail", slaveResponse.getDetail());
                finishRequestContent.put("finish", System.currentTimeMillis() / 1000);
                requestMap.put("content", new JSONObject(finishRequestContent).toString());
                String finishRequestContentReturn = ParserFastJson.requestMetadata(metadataUrls, requestMap);
                logger.info(taskId + " slave return error, update status of request, and return " + finishRequestContentReturn);
                boolean finishRequestContentFlag = ParserFastJson.returnError(finishRequestContentReturn);
                if (finishRequestContentFlag) {
                    return new DataProxyResponse(uniqKey, reqId, StatusEnum.QUERY_TASK_ERROR, finishRequestContentReturn);
                }

                Map<String, Object> finishContent = new HashMap<>();
                finishContent.put("status", "5");
                finishContent.put("detail", slaveResponse.getDetail());
                finishContent.put("finish", System.currentTimeMillis() / 1000);
                metadataMap.put("content", new JSONObject(finishContent).toString());
                String finishContentReturn = ParserFastJson.requestMetadata(metadataUrls, metadataMap);
                logger.info(taskId + " slave return error, update status of task, and return " + finishContentReturn);
                boolean finishContentFlag = ParserFastJson.returnError(finishContentReturn);
                if (finishContentFlag) {
                    return new DataProxyResponse(uniqKey, reqId, StatusEnum.QUERY_TASK_ERROR, finishContentReturn);
                }
            }
            returnMessage = "success";
            finishResponse = new DataProxyResponse(uniqKey, reqId, StatusEnum.QUERY_TASK_SUCCESS, returnMessage);
        } else {
            //5.1反馈无权限
            logger.info(taskId + " this user " + uid + " has no permissions");
            finishResponse = new DataProxyResponse(uniqKey, reqId, StatusEnum.QUERY_TASK_ERROR, returnMessage);
        }
        logger.info(taskId + " response master end");
        //8反馈任务id至数据代理接口
        return finishResponse;
    }

    @Override
    public boolean finished(TaskFinishOrderRequest taskInfo) {
        StatusEnum statusEnum = taskInfo.getStatus();
        String detail = taskInfo.getDetail();
        String task_id = taskInfo.getTask_id();
        int status = 0;
        if (statusEnum.toString().contains("SUCCESS")) {
            status = 4;
            detail = "success";
        } else if (statusEnum.toString().contains("ERROR")) {
            status = 5;
        } else if (statusEnum.toString().contains("INTERRUPTED")) {
            status = 6;
        }

        HashMap<String, String> requestMap = new HashMap<>();
        requestMap.put("scope", "request");

        HashMap<String, String> metadataMap = new HashMap<>();
        metadataMap.put("scope", "task");
        metadataMap.put("opration", "search");
        Map<String, Object> queryCondition = new HashMap<>();
        queryCondition.put("task_id", task_id);
        metadataMap.put("condition", new JSONObject(queryCondition).toString());
        metadataMap.put("content", "{\"req_id\":\"\"}");
        String queryReturn = ParserFastJson.requestMetadata(metadataUrls, metadataMap);
        logger.info(task_id + " to search req_id by task_id " + task_id + " , and return " + queryReturn);
        boolean queryFlag = ParserFastJson.returnError(queryReturn);
        if (queryFlag) {
            return false;
        }

        List<Map<String, String>> requestList = ParserFastJson.getContentByType(queryReturn, "data");
        if (!requestList.isEmpty()) {
            for (Map<String, String> map : requestList) {
                String req_ids = map.get("req_id");
                String[] req_id_arr = req_ids.split(",");
                for (String req_id_query : req_id_arr) {
                    String status_str = "";
                    requestMap.put("opration", "search");
                    Map<String, Object> statusRequestCondition = new HashMap<>();
                    statusRequestCondition.put("req_id", req_id_query);
                    requestMap.put("condition", new JSONObject(statusRequestCondition).toString());
                    requestMap.put("content", "{\"status\":\"\"}");
                    String statusRequestReturn = ParserFastJson.requestMetadata(metadataUrls, requestMap);
                    logger.info(task_id + " get status from request, and return " + statusRequestReturn);
                    boolean statusRequestFlag = ParserFastJson.returnError(statusRequestReturn);
                    if (statusRequestFlag) {
                        return false;
                    }
                    List<Map<String, String>> statusList = ParserFastJson.getContentByType(statusRequestReturn, "data");
                    if (!statusList.isEmpty()) {
                        Map<String, String> statusMap = statusList.get(0);
                        status_str = statusMap.get("status");
                        logger.debug(task_id + " this req_id " + req_id_query + " status is " + status_str);
                    } else {
                        return false;
                    }
                    if ("3".equals(status_str)) {
                        requestMap.put("opration", "update");
                        Map<String, Object> finishRequestContent = new HashMap<>();
                        finishRequestContent.put("status", status);
                        finishRequestContent.put("detail", detail);
                        finishRequestContent.put("finish", System.currentTimeMillis() / 1000);
                        requestMap.put("content", new JSONObject(finishRequestContent).toString());
                        Map<String, Object> finishRequestCondition = new HashMap<>();
                        finishRequestCondition.put("req_id", req_id_query);
                        requestMap.put("condition", new JSONObject(finishRequestCondition).toString());
                        String finishContentReturn = ParserFastJson.requestMetadata(metadataUrls, requestMap);
                        logger.info(task_id + " slave done, update status of request, req_id is " + req_id_query + ", and return " + finishContentReturn);
                        boolean finishContentFlag = ParserFastJson.returnError(finishContentReturn);
                        if (finishContentFlag) {
                            return false;
                        }
                    } else {
                        logger.info(task_id + " finish, can't update status of request, cause req_id is " + req_id_query + ", and status is " + status_str);
                    }
                }
            }
        }

        metadataMap.put("opration", "update");
        Map<String, Object> finishContent = new HashMap<>();
        finishContent.put("status", status);
        finishContent.put("detail", detail);
        finishContent.put("finish", System.currentTimeMillis() / 1000);
        metadataMap.put("content", new JSONObject(finishContent).toString());
        Map<String, Object> finishCondition = new HashMap<>();
        finishCondition.put("task_id", task_id);
        metadataMap.put("condition", new JSONObject(finishCondition).toString());
        String finishContentReturn = ParserFastJson.requestMetadata(metadataUrls, metadataMap);
        logger.info(task_id + " slave done, update status of task, and return " + finishContentReturn);
        boolean finishContentFlag = ParserFastJson.returnError(finishContentReturn);
        return !finishContentFlag;
    }

    @Override
    public DataProxyResponse redo(TaskInfo taskInfo) {
        String timestamp = TIMEFORMAT.format(new Date(System.currentTimeMillis()));
        DataProxyResponse finishResponse = null;
        String task_id = taskInfo.getTask_id();
        String taskResult = taskInfo.getResult_path();
        String sql = taskInfo.getSql();
        ScopeEnum scopeEnum = taskInfo.getScope();
        int count = RequestCount.addUp();
        String uniq_key = "redo_" + task_id + "_" + timestamp + "_" + count;
        TaskTypeEnum typeEnum = TaskTypeEnum.RUN;
        String req_id = "";
        logger.info(task_id + " REDO, the sql is " + sql + " ,scope is " + scopeEnum.toString() + " ,uniq_key is " + uniq_key);

        HashMap<String, String> metadataMap = new HashMap<>();
        metadataMap.put("scope", "task");
        metadataMap.put("opration", "search");
        Map<String, Object> queryCondition = new HashMap<>();
        queryCondition.put("task_id", task_id);
        metadataMap.put("condition", new JSONObject(queryCondition).toString());
        metadataMap.put("content", "{\"req_id\":\"\"}");
        String queryReturn = ParserFastJson.requestMetadata(metadataUrls, metadataMap);
        logger.info(task_id + " to search req_id by task_id " + task_id + " , and return " + queryReturn);
        boolean queryFlag = ParserFastJson.returnError(queryReturn);
        if (queryFlag) {
            return new DataProxyResponse(uniq_key, req_id, StatusEnum.QUERY_TASK_ERROR, queryReturn);
        }

        List<Map<String, String>> requestList = ParserFastJson.getContentByType(queryReturn, "data");
        if (!requestList.isEmpty()) {
            Map<String, String> taskMap = requestList.get(0);
            req_id = taskMap.get("req_id");

            metadataMap.put("opration", "update");
            Map<String, Object> queryTaskCondition = new HashMap<>();
            queryTaskCondition.put("task_id", task_id);
            metadataMap.put("condition", new JSONObject(queryTaskCondition).toString());

            HashMap<String, String> requestMap = new HashMap<>();
            requestMap.put("scope", "request");
            requestMap.put("opration", "update");
            Map<String, Object> updateRequestStatus = new HashMap<>();
            updateRequestStatus.put("status", "3");
            requestMap.put("content", new JSONObject(updateRequestStatus).toString());
            //9给slave分派任务
            HashMap<String, Object> salveMap = new HashMap<>();
            salveMap.put("sql", sql);
            salveMap.put("scope", scopeEnum);
            salveMap.put("uniq_key", uniq_key);
            salveMap.put("type", typeEnum);
            salveMap.put("result_path", taskResult);
            salveMap.put("task_id", task_id);
            salveMap.put("uid", "");
            TaskOrderResponse slaveResponse = SlaveTools.sendToSlave(producer, salveMap);
            StatusEnum slaveEnum = slaveResponse.getStatus();
            logger.info(task_id + " REDO, allocating task to salve, and return " + slaveEnum);
            if (slaveEnum.equals(StatusEnum.RUNNING)) {
                //10更新任务状态
                for (Map<String, String> map : requestList) {
                    String req_id_query = map.get("req_id");
                    Map<String, Object> queryRequestCondition = new HashMap<>();
                    queryRequestCondition.put("req_id", req_id_query);
                    requestMap.put("condition", new JSONObject(queryRequestCondition).toString());
                    String updateRequestStatusReturn = ParserFastJson.requestMetadata(metadataUrls, requestMap);
                    logger.info(task_id + " slave begin to run, update status of request, and return " + updateRequestStatusReturn);
                    boolean updateRequestStatusFlag = ParserFastJson.returnError(updateRequestStatusReturn);
                    if (updateRequestStatusFlag) {
                        return new DataProxyResponse(uniq_key, req_id, StatusEnum.QUERY_TASK_ERROR, updateRequestStatusReturn);
                    }
                }

                Map<String, Object> updateStatus = new HashMap<>();
                updateStatus.put("status", "3");
                metadataMap.put("content", new JSONObject(updateStatus).toString());
                String updateStatusReturn = ParserFastJson.requestMetadata(metadataUrls, metadataMap);
                logger.info(task_id + " REDO, slave begin to run, update status of task, and return " + updateStatusReturn);
                boolean updateStatusFlag = ParserFastJson.returnError(updateStatusReturn);
                if (updateStatusFlag) {
                    return new DataProxyResponse(uniq_key, req_id, StatusEnum.QUERY_TASK_ERROR, updateStatusReturn);
                }
                finishResponse = new DataProxyResponse(uniq_key, req_id, StatusEnum.QUERY_TASK_SUCCESS, "success");
            } else {
                for (Map<String, String> map : requestList) {
                    String req_id_query = map.get("req_id");
                    Map<String, Object> finishRequestCondition = new HashMap<>();
                    finishRequestCondition.put("req_id", req_id_query);
                    requestMap.put("condition", new JSONObject(finishRequestCondition).toString());
                    Map<String, Object> finishRequestContent = new HashMap<>();
                    finishRequestContent.put("status", "5");
                    finishRequestContent.put("detail", slaveResponse.getDetail());
                    finishRequestContent.put("finish", System.currentTimeMillis() / 1000);
                    requestMap.put("content", new JSONObject(finishRequestContent).toString());
                    String finishRequestContentReturn = ParserFastJson.requestMetadata(metadataUrls, requestMap);
                    logger.info(task_id + " REDO, slave return error, update status ofrequest, and return " + finishRequestContentReturn);
                    boolean finishRequestContentFlag = ParserFastJson.returnError(finishRequestContentReturn);
                    if (finishRequestContentFlag) {
                        return new DataProxyResponse(uniq_key, req_id, StatusEnum.QUERY_TASK_ERROR, finishRequestContentReturn);
                    }
                }

                Map<String, Object> finishContent = new HashMap<>();
                finishContent.put("status", "5");
                finishContent.put("detail", slaveResponse.getDetail());
                finishContent.put("finish", System.currentTimeMillis() / 1000);
                metadataMap.put("content", new JSONObject(finishContent).toString());
                String finishContentReturn = ParserFastJson.requestMetadata(metadataUrls, metadataMap);
                logger.info(task_id + " REDO, slave return error, update status of task, and return " + finishContentReturn);
                boolean finishContentFlag = ParserFastJson.returnError(finishContentReturn);
                if (finishContentFlag) {
                    return new DataProxyResponse(uniq_key, req_id, StatusEnum.QUERY_TASK_ERROR, finishContentReturn);
                }
                finishResponse = new DataProxyResponse(uniq_key, req_id, StatusEnum.QUERY_TASK_ERROR, slaveResponse.getDetail());
            }
        } else {
            logger.info(task_id + " REDO, search req_id by task_id, return empty");
            finishResponse = new DataProxyResponse(uniq_key, req_id, StatusEnum.QUERY_TASK_ERROR, queryReturn);
        }
        logger.info(task_id + " REDO, response master end");
        return finishResponse;
    }

    @Override
    public String getName() {
        throw new UnsupportedOperationException("Not supported yet."); //To change body of generated methods, choose Tools | Templates.
    }

    @Override
    public void stop() {
        throw new UnsupportedOperationException("Not supported yet."); //To change body of generated methods, choose Tools | Templates.
    }
//
//    private Boolean queryTaskFull(String uid) throws Exception {
//        String uids;
//        JSONObject uidResponse = getUidsInAgency(uid);
//        JSONObject result = uidResponse.getJSONObject("result");
//        if ("success".equals(uidResponse.getString("status"))) {
//            uids = result.getString("uids");
//            uids = uids.replace("[", "(").replace("]", ")");
//        } else {
//            logger.error(result);
//            throw new Exception(result.toJSONString());
//
//        }
//
//        int queryTaskNum = 0;
//        JSONObject queryTaskResponse = getQueryTaskNum(uids);
//        if ("succeeded".equals(queryTaskResponse.getString("status"))) {
//            JSONObject queryTaskData = JSONObject.parseObject(String.valueOf(queryTaskResponse.getJSONArray("data").get(0)));
//            queryTaskNum = queryTaskData.getIntValue("count(*)");
//        } else {
//            logger.error(queryTaskResponse.getString("detail"));
//            throw new Exception(queryTaskResponse.getString("detail"));
//        }
//        return queryTaskNum > queryMaxNum;
//    }
//
//    private JSONObject getUidsInAgency(String uid) throws Exception {
//        List<NameValuePair> params = new ArrayList<>();
//        params.add(new BasicNameValuePair("uid", uid));
//
//        String response;
//        try {
//            response = HttpHelper.httpPost("http://192.168.11.187:8000/get_all_uid/", params);
//        } catch (Exception e) {
//            logger.error(ExceptionUtils.getFullStackTrace(e));
//            throw new Exception("get sql type media http error.");
//        }
//
//        return JSONObject.parseObject(response);
//    }
//
//    private JSONObject getQueryTaskNum(String url, String uids, String type) throws Exception {
//        long timestampNow = System.currentTimeMillis() / 1000;
//        long timestampStart = timestampNow - 24 * 60 * 60;
//        String sql = "select count(*) from msg_db.request where start>" + timestampStart + " and type=" + type + " and uid in " + uids;
//        List<NameValuePair> params = new ArrayList<>();
//        params.add(new BasicNameValuePair("scope", "audit"));
//        params.add(new BasicNameValuePair("opration", "search"));
//        params.add(new BasicNameValuePair("condition", sql));
//
//        String response;
//        try {
//            response = HttpHelper.httpPost(url, params);
//        } catch (Exception e) {
//            logger.error(ExceptionUtils.getFullStackTrace(e));
//            throw new Exception("get sql type media http error.");
//        }
//
//        return JSONObject.parseObject(response);
//    }
}
